package net.qiqbframework.config;


import net.qiqbframework.commandhandling.MessageHandlerInterceptorSupport;
import net.qiqbframework.common.transaction.NoTransactionManager;
import net.qiqbframework.common.transaction.TransactionManager;
import net.qiqbframework.eventhandling.*;
import net.qiqbframework.messaging.Message;
import net.qiqbframework.messaging.MessageHandlerInterceptor;
import net.qiqbframework.messaging.SubscribableMessageSource;
import net.qiqbframework.messaging.annotation.MessageHandlingMemberFactory;
import net.qiqbframework.messaging.intercepting.CorrelationDataInterceptor;
import net.qiqbframework.messaging.monitoring.MessageMonitor;
import net.qiqbframework.messaging.task.RollbackConfiguration;
import net.qiqbframework.messaging.task.RollbackConfigurationType;

import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;

import static java.lang.String.format;
import static java.util.Comparator.comparing;
import static net.qiqbframework.common.BuilderUtils.assertNonNull;
import static net.qiqbframework.common.utils.AnnotationUtil.findAnnotationAttributes;

public class EventProcessingModule implements ModuleConfiguration, EventProcessingConfiguration, EventProcessingConfigurer {

    private static final String CONFIGURED_DEFAULT_TEP_CONFIG = "___DEFAULT_TEP_CONFIG";

    private static final Function<Class<?>, String> DEFAULT_SAGA_PROCESSING_GROUP_FUNCTION =
            c -> c.getSimpleName() + "Processor";

    private final List<TypeProcessingGroupSelector> typeSelectors = new ArrayList<>();
    private final List<InstanceProcessingGroupSelector> instanceSelectors = new ArrayList<>();
    private final Map<String, String> processingGroupsAssignments = new HashMap<>();
    // the default selector determines the processing group by inspecting the @ProcessingGroup annotation
    private final TypeProcessingGroupSelector annotationGroupSelector = TypeProcessingGroupSelector
            .defaultSelector(type -> annotatedProcessingGroupOfType(type).orElse(null));
    private TypeProcessingGroupSelector typeFallback =
            TypeProcessingGroupSelector.defaultSelector(DEFAULT_SAGA_PROCESSING_GROUP_FUNCTION);
    private InstanceProcessingGroupSelector instanceFallbackSelector = InstanceProcessingGroupSelector.defaultSelector(EventProcessingModule::packageOfObject);

    private final List<ComponentHolder<Object>> eventHandlerBuilders = new ArrayList<>();
    private final Map<String, EventProcessorBuilder> eventProcessorBuilders = new HashMap<>();

    protected final Map<String, ComponentHolder<EventProcessor>> eventProcessors = new HashMap<>();

    protected final List<BiFunction<Configuration, String, MessageHandlerInterceptor<? super EventMessage<?>>>> defaultHandlerInterceptors = new ArrayList<>();
    protected final Map<String, List<Function<Configuration, MessageHandlerInterceptor<? super EventMessage<?>>>>> handlerInterceptorsBuilders = new HashMap<>();
    protected final Map<String, ComponentHolder<ListenerInvocationErrorHandler>> listenerInvocationErrorHandlers = new HashMap<>();
    protected final Map<String, ComponentHolder<ErrorHandler>> errorHandlers = new HashMap<>();
    protected final Map<String, MessageMonitorFactory> messageMonitorFactories = new HashMap<>();
    protected final Map<String, ComponentHolder<RollbackConfiguration>> rollbackConfigurations = new HashMap<>();
    protected final Map<String, ComponentHolder<TransactionManager>> transactionManagers = new HashMap<>();

    protected final Map<String, ComponentHolder<TrackingEventProcessorConfiguration>> tepConfigs = new HashMap<>();

    private final AtomicBoolean initialized = new AtomicBoolean(false);
    protected Configuration configuration;

    private final ComponentHolder<ListenerInvocationErrorHandler> defaultListenerInvocationErrorHandler = new ComponentHolder<>(
            () -> configuration,
            "listenerInvocationErrorHandler",
            c -> c.getComponent(ListenerInvocationErrorHandler.class, LoggingErrorHandler::new)
    );
    private final ComponentHolder<ErrorHandler> defaultErrorHandler = new ComponentHolder<>(
            () -> configuration,
            "errorHandler",
            c -> c.getComponent(ErrorHandler.class, PropagatingErrorHandler::instance)
    );

    private final ComponentHolder<RollbackConfiguration> defaultRollbackConfiguration = new ComponentHolder<>(
            () -> configuration,
            "rollbackConfiguration",
            c -> c.getComponent(RollbackConfiguration.class, () -> RollbackConfigurationType.ANY_THROWABLE));

    private final ComponentHolder<TransactionManager> defaultTransactionManager = new ComponentHolder<>(
            () -> configuration,
            "transactionManager",
            c -> c.getComponent(TransactionManager.class, NoTransactionManager::instance)
    );


    private final ComponentHolder<SubscribableMessageSource<EventMessage<?>>> defaultSubscribableSource =
            new ComponentHolder<>(
                    () -> configuration,
                    "defaultSubscribableMessageSource",
                    Configuration::eventBus
            );
    private final ComponentHolder<TrackingEventProcessorConfiguration> defaultTepConfig =
            new ComponentHolder<>(
                    () -> configuration,
                    "trackingEventProcessorConfiguration",
                    c -> c.getComponent(
                            TrackingEventProcessorConfiguration.class,
                            TrackingEventProcessorConfiguration::forSingleThreadedProcessing
                    )
            );
    private EventProcessorBuilder defaultEventProcessorBuilder = this::defaultEventProcessor;
    private Function<String, String> defaultProcessingGroupAssignment = Function.identity();

    @Override
    public void initialize(Configuration configuration) {
        this.configuration = configuration;
        eventProcessors.clear();
        configuration.onStart(Integer.MIN_VALUE, this::initializeProcessors);
    }

    /**
     * Initializes the event processors by assigning each of the event handlers according to the defined selectors. When
     * processors have already been initialized, this method does nothing.
     */
    private void initializeProcessors() {
        if (initialized.get()) {
            return;
        }

        synchronized (initialized) {
            if (initialized.get()) {
                return;
            }

            instanceSelectors.sort(comparing(InstanceProcessingGroupSelector::getPriority).reversed());

            Map<String, List<Function<Configuration, EventMessageHandler>>> handlerInvokers = new HashMap<>();
            registerEventHandlerInvokers(handlerInvokers);


            handlerInvokers.forEach((processorName, invokers) -> {
                ComponentHolder<EventProcessor> eventProcessorComponent = new ComponentHolder<>(
                        configuration, processorName, c -> buildEventProcessor(invokers, processorName)
                );
                eventProcessors.put(processorName, eventProcessorComponent);
            });

            eventProcessors.values().forEach(ComponentHolder::get);
            initialized.set(true);
        }
    }

    private String selectProcessingGroupByType(Class<?> type) {
        // when selecting on type,
        List<TypeProcessingGroupSelector> selectors = new ArrayList<>(typeSelectors);
        selectors.add(annotationGroupSelector);
        selectors.add(typeFallback);

        return selectors.stream()
                .map(s -> s.select(type))
                .filter(Optional::isPresent)
                .map(Optional::get)
                .findFirst()
                .orElseThrow(() -> new IllegalStateException(
                        "Could not select a processing group for type [" + type.getSimpleName() + "]"
                ));
    }

    private void registerEventHandlerInvokers(
            Map<String, List<Function<Configuration, EventMessageHandler>>> handlerInvokers
    ) {
        Map<String, List<Object>> assignments = new HashMap<>();

        // we combine the selectors in the order of precedence (instances, then types, then default instance, default types and fallbacks)
        List<InstanceProcessingGroupSelector> selectors = new ArrayList<>(instanceSelectors);
        typeSelectors.stream().map(InstanceToTypeProcessingGroupSelectorAdapter::new).forEach(selectors::add);
        selectors.add(new InstanceToTypeProcessingGroupSelectorAdapter(annotationGroupSelector));
        selectors.add(instanceFallbackSelector);

        eventHandlerBuilders.stream()
                .map(ComponentHolder::get)
                .forEach(handler -> {
                    String processingGroup =
                            selectors.stream()
                                    .map(s -> s.select(handler))
                                    .filter(Optional::isPresent)
                                    .map(Optional::get)
                                    .findFirst()
                                    .orElseThrow(() -> new IllegalStateException(
                                            "Could not select a processing group for handler ["
                                                    + handler.getClass().getSimpleName() + "]"
                                    ));
                    assignments.computeIfAbsent(processingGroup, k -> new ArrayList<>())
                            .add(handler);
                });
        assignments.forEach((processingGroup, handlers) -> {
            String processorName = processorNameForProcessingGroup(processingGroup);

            handlerInvokers.computeIfAbsent(processorName, k -> new ArrayList<>())
                    .add(
                            c -> simpleInvoker(processingGroup, handlers)
                    );
        });
    }

    private EventMessageHandler simpleInvoker(String processingGroup, List<Object> handlers) {
        return SimpleEventHandlerInvoker.builder()
                .eventHandlers(handlers)
                .handlerDefinition(retrieveHandlerDefinition(handlers))
                .parameterResolverFactory(configuration.parameterResolverFactory())
                .listenerInvocationErrorHandler(listenerInvocationErrorHandler(processingGroup))

                .build();
    }

    public ListenerInvocationErrorHandler listenerInvocationErrorHandler(String processingGroup) {
        validateConfigInitialization();
        return listenerInvocationErrorHandlers.containsKey(processingGroup)
                ? listenerInvocationErrorHandlers.get(processingGroup).get()
                : defaultListenerInvocationErrorHandler.get();
    }

    private MessageHandlingMemberFactory retrieveHandlerDefinition(List<Object> handlers) {
        return configuration.handlerDefinition(handlers.get(0).getClass());
    }


    private EventProcessor buildEventProcessor(List<Function<Configuration, EventMessageHandler>> builderFunctions,
                                               String processorName) {
        List<EventMessageHandler> invokers = builderFunctions
                .stream()
                .map(invokerBuilder -> invokerBuilder.apply(configuration))
                .collect(Collectors.toList());
        EventMessageHandler multiEventHandlerInvoker = new MultiEventHandlerInvoker(invokers);

        EventProcessor eventProcessor = eventProcessorBuilders
                .getOrDefault(processorName, defaultEventProcessorBuilder)
                .build(processorName, configuration, multiEventHandlerInvoker);

        addInterceptors(processorName, eventProcessor);

        return eventProcessor;
    }

    private void addInterceptors(String processorName, MessageHandlerInterceptorSupport<EventMessage<?>> processor) {
        handlerInterceptorsBuilders.getOrDefault(processorName, new ArrayList<>())
                .stream()
                .map(hi -> hi.apply(configuration))
                .forEach(processor::registerHandlerInterceptor);

        defaultHandlerInterceptors.stream()
                .map(f -> f.apply(configuration, processorName))
                .filter(Objects::nonNull)
                .forEach(processor::registerHandlerInterceptor);

        processor.registerHandlerInterceptor(new CorrelationDataInterceptor<>(configuration.correlationDataProviders()));
    }

    //<editor-fold desc="configuration methods">
    @SuppressWarnings("unchecked")
    @Override
    public <T extends EventProcessor> Optional<T> eventProcessorByProcessingGroup(String processingGroup) {
        return Optional.ofNullable((T) eventProcessors().get(processorNameForProcessingGroup(processingGroup)));
    }

    @Override
    public Map<String, EventProcessor> eventProcessors() {
        validateConfigInitialization();
        initializeProcessors();
        Map<String, EventProcessor> result = new HashMap<>(eventProcessors.size());
        eventProcessors.forEach((name, component) -> result.put(name, component.get()));
        return result;
    }

    @Override
    public List<MessageHandlerInterceptor<? super EventMessage<?>>> interceptorsFor(String processorName) {
        validateConfigInitialization();
        return eventProcessor(processorName).map(EventProcessor::getHandlerInterceptors)
                .orElse(Collections.emptyList());
    }


    @Override
    public RollbackConfiguration rollbackConfiguration(String processorName) {
        validateConfigInitialization();
        return rollbackConfigurations.containsKey(processorName)
                ? rollbackConfigurations.get(processorName).get()
                : defaultRollbackConfiguration.get();
    }

    @Override
    public ErrorHandler errorHandler(String processorName) {
        validateConfigInitialization();
        return errorHandlers.containsKey(processorName)
                ? errorHandlers.get(processorName).get()
                : defaultErrorHandler.get();
    }


    private String processorNameForProcessingGroup(String processingGroup) {
        validateConfigInitialization();
        return processingGroupsAssignments.getOrDefault(processingGroup,
                defaultProcessingGroupAssignment
                        .apply(processingGroup));
    }

    @Override
    public MessageMonitor<? super Message<?>> messageMonitor(Class<?> componentType,
                                                             String eventProcessorName) {
        validateConfigInitialization();
        if (messageMonitorFactories.containsKey(eventProcessorName)) {
            return messageMonitorFactories.get(eventProcessorName).create(configuration,
                    componentType,
                    eventProcessorName);
        } else {
            return configuration.messageMonitor(componentType, eventProcessorName);
        }
    }


    @Override
    public TransactionManager transactionManager(String processorName) {
        validateConfigInitialization();
        return transactionManagers.containsKey(processorName)
                ? transactionManagers.get(processorName).get()
                : defaultTransactionManager.get();
    }


    private void validateConfigInitialization() {
        assertNonNull(
                configuration, "Cannot proceed because the Configuration is not initialized for this module yet."
        );
    }


    @Override
    public EventProcessingConfigurer registerEventHandler(
            Function<Configuration, Object> eventHandlerBuilder) {
        this.eventHandlerBuilders.add(new ComponentHolder<>(() -> configuration,
                "eventHandler",
                eventHandlerBuilder));
        return this;
    }

    //</editor-fold>

    @Override
    public EventProcessingConfigurer registerDefaultListenerInvocationErrorHandler(
            Function<Configuration, ListenerInvocationErrorHandler> listenerInvocationErrorHandlerBuilder) {
        defaultListenerInvocationErrorHandler.update(listenerInvocationErrorHandlerBuilder);
        return this;
    }

    @Override
    public EventProcessingConfigurer registerListenerInvocationErrorHandler(String processingGroup,
                                                                            Function<Configuration, ListenerInvocationErrorHandler> listenerInvocationErrorHandlerBuilder) {
        listenerInvocationErrorHandlers.put(processingGroup, new ComponentHolder<>(() -> configuration,
                "listenerInvocationErrorHandler",
                listenerInvocationErrorHandlerBuilder));
        return this;
    }



    @Override
    public EventProcessingConfigurer configureDefaultSubscribableMessageSource(Function<Configuration, SubscribableMessageSource<EventMessage<?>>> defaultSource) {
        this.defaultSubscribableSource.update(defaultSource);
        return this;
    }


    @Override
    public EventProcessingConfigurer registerEventProcessorFactory(
            EventProcessorBuilder eventProcessorBuilder) {
        this.defaultEventProcessorBuilder = eventProcessorBuilder;
        return this;
    }

    @Override
    public EventProcessingConfigurer registerEventProcessor(String name,
                                                            EventProcessorBuilder eventProcessorBuilder) {
        if (this.eventProcessorBuilders.containsKey(name)) {
            throw new ConfigurationException(format("Event processor with name %s already exists", name));
        }
        this.eventProcessorBuilders.put(name, eventProcessorBuilder);
        return this;
    }


    @Override
    public EventProcessingConfigurer usingSubscribingEventProcessors() {
        this.defaultEventProcessorBuilder = (name, conf, eventHandlerInvoker) ->
                subscribingEventProcessor(name, eventHandlerInvoker, defaultSubscribableSource.get());
        return this;
    }


    @Override
    public EventProcessingConfigurer registerSubscribingEventProcessor(String name,
                                                                       Function<Configuration, SubscribableMessageSource<EventMessage<?>>> messageSource) {
        registerEventProcessor(name, (n, c, ehi) -> subscribingEventProcessor(n, ehi, messageSource.apply(c)));
        return this;
    }

    @Override
    public EventProcessingConfigurer registerDefaultErrorHandler(
            Function<Configuration, ErrorHandler> errorHandlerBuilder) {
        this.defaultErrorHandler.update(errorHandlerBuilder);
        return this;
    }

    @Override
    public EventProcessingConfigurer registerErrorHandler(String eventProcessorName,
                                                          Function<Configuration, ErrorHandler> errorHandlerBuilder) {
        this.errorHandlers.put(eventProcessorName, new ComponentHolder<>(() -> configuration,
                "errorHandler",
                errorHandlerBuilder));
        return this;
    }

    @Override
    public EventProcessingConfigurer byDefaultAssignHandlerInstancesTo(Function<Object, String> assignmentFunction) {
        this.instanceFallbackSelector = InstanceProcessingGroupSelector.defaultSelector(assignmentFunction);
        return this;
    }

    @Override
    public EventProcessingConfigurer byDefaultAssignHandlerTypesTo(Function<Class<?>, String> assignmentFunction) {
        this.typeFallback = TypeProcessingGroupSelector.defaultSelector(assignmentFunction);
        return this;
    }

    @Override
    public EventProcessingConfigurer assignHandlerInstancesMatching(String processingGroup, int priority,
                                                                    Predicate<Object> criteria) {
        this.instanceSelectors.add(new InstanceProcessingGroupSelector(processingGroup, priority, criteria));
        return this;
    }

    @Override
    public EventProcessingConfigurer assignHandlerTypesMatching(String processingGroup, int priority,
                                                                Predicate<Class<?>> criteria) {
        this.typeSelectors.add(new TypeProcessingGroupSelector(processingGroup, priority, criteria));
        return this;
    }

    @Override
    public EventProcessingConfigurer assignProcessingGroup(String processingGroup, String processorName) {
        this.processingGroupsAssignments.put(processingGroup, processorName);
        return this;
    }

    @Override
    public EventProcessingConfigurer assignProcessingGroup(Function<String, String> assignmentRule) {
        this.defaultProcessingGroupAssignment = assignmentRule;
        return this;
    }

    @Override
    public EventProcessingConfigurer registerHandlerInterceptor(String processorName,
                                                                Function<Configuration, MessageHandlerInterceptor<? super EventMessage<?>>> interceptorBuilder) {
        ComponentHolder<EventProcessor> eps = eventProcessors.get(processorName);
        if (eps != null && eps.isInitialized()) {
            eps.get().registerHandlerInterceptor(interceptorBuilder.apply(configuration));
        }
        this.handlerInterceptorsBuilders.computeIfAbsent(processorName, k -> new ArrayList<>())
                .add(interceptorBuilder);
        return this;
    }

    @Override
    public EventProcessingConfigurer registerDefaultHandlerInterceptor(
            BiFunction<Configuration, String, MessageHandlerInterceptor<? super EventMessage<?>>> interceptorBuilder) {
        this.defaultHandlerInterceptors.add(interceptorBuilder);
        return this;
    }


    @Override
    public EventProcessingConfigurer registerMessageMonitorFactory(String eventProcessorName,
                                                                   MessageMonitorFactory messageMonitorFactory) {
        this.messageMonitorFactories.put(eventProcessorName, messageMonitorFactory);
        return this;
    }

    @Override
    public EventProcessingConfigurer registerRollbackConfiguration(String name,
                                                                   Function<Configuration, RollbackConfiguration> rollbackConfigurationBuilder) {
        this.rollbackConfigurations.put(name, new ComponentHolder<>(() -> configuration,
                "rollbackConfiguration",
                rollbackConfigurationBuilder));
        return this;
    }

    @Override
    public EventProcessingConfigurer registerTransactionManager(String name,
                                                                Function<Configuration, TransactionManager> transactionManagerBuilder) {
        this.transactionManagers.put(name, new ComponentHolder<>(() -> configuration,
                "transactionManager",
                transactionManagerBuilder));
        return this;
    }

    @Override
    public EventProcessingConfigurer registerDefaultTransactionManager(
            Function<Configuration, TransactionManager> transactionManagerBuilder
    ) {
        this.defaultTransactionManager.update(transactionManagerBuilder);
        return this;
    }

    @Override
    public EventProcessingConfigurer registerTrackingEventProcessorConfiguration(
            String name,
            Function<Configuration, TrackingEventProcessorConfiguration> trackingEventProcessorConfigurationBuilder
    ) {
        this.tepConfigs.put(name, new ComponentHolder<>(() -> configuration,
                "trackingEventProcessorConfiguration",
                trackingEventProcessorConfigurationBuilder));
        return this;
    }

    @Override
    public EventProcessingConfigurer registerTrackingEventProcessorConfiguration(
            Function<Configuration, TrackingEventProcessorConfiguration> trackingEventProcessorConfigurationBuilder
    ) {
        this.tepConfigs.put(CONFIGURED_DEFAULT_TEP_CONFIG,
                new ComponentHolder<>(() -> configuration,
                        "trackingEventProcessorConfiguration",
                        trackingEventProcessorConfigurationBuilder));
        return this;
    }


    private EventProcessor defaultEventProcessor(String name,
                                                 Configuration conf,
                                                 EventMessageHandler eventHandlerInvoker) {
        return subscribingEventProcessor(name, eventHandlerInvoker, defaultSubscribableSource.get());
    }

    private TrackingEventProcessorConfiguration trackingEventProcessorConfig(String name) {
        if (tepConfigs.containsKey(name)) {
            return tepConfigs.get(name).get();
        }
        return tepConfigs.getOrDefault(CONFIGURED_DEFAULT_TEP_CONFIG, defaultTepConfig).get();
    }

    /**
     * Default {@link SubscribingEventProcessor} configuration based on this configure module.
     *
     * @param name                of the processor
     * @param eventHandlerInvoker used by the processor for the vent handling
     * @param messageSource       where to retrieve events from
     * @return Default {@link SubscribingEventProcessor} configuration based on this configure module.
     */
    protected EventProcessor subscribingEventProcessor(String name,
                                                       EventMessageHandler eventHandlerInvoker,
                                                       SubscribableMessageSource<EventMessage<?>> messageSource) {
        return SubscribingEventProcessorBuilder.getInstance()
                .name(name)
                .eventHandlerInvoker(eventHandlerInvoker)
                .rollbackConfiguration(rollbackConfiguration(name))
                .errorHandler(errorHandler(name))
                .messageMonitor(messageMonitor(SubscribingEventProcessor.class, name))
                .messageSource(messageSource)
                .processingStrategy(DirectEventProcessingStrategy.INSTANCE)
                .transactionManager(transactionManager(name))
                .spanFactory(configuration.getComponent(EventProcessorSpanFactory.class))
                .build();
    }


    /**
     * Gets the package name from the class of the given object.
     * <p>
     * Since class.getPackage() can be null e.g. for generated classes, the package name is determined the old fashioned
     * way based on the full qualified class name.
     *
     * @param object {@link Object}
     * @return {@link String}
     */
    protected static String packageOfObject(Object object) {
        return object.getClass().getName().replace("." + object.getClass().getSimpleName(), "");
    }

    private static Optional<String> annotatedProcessingGroupOfType(Class<?> type) {
        Optional<Map<String, Object>> annAttr = findAnnotationAttributes(type, ProcessingGroup.class);
        return annAttr.map(attr -> (String) attr.get("processingGroup"));
    }

    //<editor-fold desc="configuration state">
    private static class InstanceProcessingGroupSelector extends ProcessingGroupSelector<Object> {

        private static InstanceProcessingGroupSelector defaultSelector(Function<Object, String> selectorFunction) {
            return new InstanceProcessingGroupSelector(Integer.MIN_VALUE, selectorFunction.andThen(Optional::ofNullable));
        }

        private InstanceProcessingGroupSelector(int priority, Function<Object, Optional<String>> selectorFunction) {
            super(priority, selectorFunction);
        }

        private InstanceProcessingGroupSelector(String name, int priority, Predicate<Object> criteria) {
            super(name, priority, criteria);
        }
    }

    private static class TypeProcessingGroupSelector extends ProcessingGroupSelector<Class<?>> {

        private static TypeProcessingGroupSelector defaultSelector(Function<Class<?>, String> selectorFunction) {
            return new TypeProcessingGroupSelector(Integer.MIN_VALUE, selectorFunction.andThen(Optional::ofNullable));
        }

        private TypeProcessingGroupSelector(int priority, Function<Class<?>, Optional<String>> selectorFunction) {
            super(priority, selectorFunction);
        }

        private TypeProcessingGroupSelector(String name, int priority, Predicate<Class<?>> criteria) {
            super(name, priority, criteria);
        }
    }

    private static class InstanceToTypeProcessingGroupSelectorAdapter extends InstanceProcessingGroupSelector {

        private InstanceToTypeProcessingGroupSelectorAdapter(TypeProcessingGroupSelector delegate) {
            super(delegate.getPriority(), i -> delegate.select(i.getClass()));
        }
    }

    private static class ProcessingGroupSelector<T> {

        private final int priority;
        private final Function<T, Optional<String>> function;


        private ProcessingGroupSelector(int priority, Function<T, Optional<String>> selectorFunction) {
            this.priority = priority;
            this.function = selectorFunction;
        }

        private ProcessingGroupSelector(String name, int priority, Predicate<T> criteria) {
            this(priority, handler -> {
                if (criteria.test(handler)) {
                    return Optional.of(name);
                }
                return Optional.empty();
            });
        }

        public Optional<String> select(T handler) {
            return function.apply(handler);
        }

        public int getPriority() {
            return priority;
        }
    }
    //</editor-fold>
}
